package site.anxuan.seckill.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import site.anxuan.seckill.dao.ItemDao;
import site.anxuan.seckill.dao.ItemStockDao;
import site.anxuan.seckill.dao.PromotionDao;
import site.anxuan.seckill.entity.Item;
import site.anxuan.seckill.entity.ItemStock;
import site.anxuan.seckill.entity.Promotion;
import site.anxuan.seckill.service.ItemService;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class ItemServiceImpl implements ItemService {

    private final Logger logger = LoggerFactory.getLogger(ItemServiceImpl.class);

    private final ItemDao itemDao;

    private final ItemStockDao itemStockDao;

    private final PromotionDao promotionDao;

    private final RedisTemplate<String, Object> redisTemplate;

    private final RocketMQTemplate rocketMQTemplate;

    private Cache<String, Object> guavaCache;

    @Autowired
    public ItemServiceImpl(ItemDao itemDao, ItemStockDao itemStockDao, PromotionDao promotionDao,
                           RedisTemplate<String, Object> redisTemplate, RocketMQTemplate rocketMQTemplate) {
        this.itemDao = itemDao;
        this.itemStockDao = itemStockDao;
        this.promotionDao = promotionDao;
        this.redisTemplate = redisTemplate;
        this.rocketMQTemplate = rocketMQTemplate;
    }

    @PostConstruct
    public void init() {
        guavaCache = CacheBuilder.newBuilder()
                .initialCapacity(10)
                .maximumSize(100)
                .expireAfterWrite(1, TimeUnit.MINUTES)
                .build();
    }

    @Override
    public List<Item> getListOnPromotion() {
        List<Item> items = itemDao.getListOnPromotion();
        return items.stream().peek(item -> {
            ItemStock itemStock = itemStockDao.getByItemId(item.getId());
            Promotion promotion = promotionDao.getByItemId(item.getId());
            item.setItemStock(itemStock);
            if (promotion != null && promotion.getStatus() == 0) {
                item.setPromotion(promotion);
            }
        }).collect(Collectors.toList());
    }

    @Override
    public Item getById(Integer itemId) {
        Item item = itemDao.getById(itemId);
        if (item == null) {
            // 缓存空对象
            Item emptyItem = new Item();
            guavaCache.put("item:" + itemId, emptyItem);
            redisTemplate.opsForValue().set("item:" + itemId, emptyItem, 3 + new Random().nextInt(8), TimeUnit.MINUTES);
            return null;
        }
        ItemStock itemStock = itemStockDao.getByItemId(item.getId());
        Promotion promotion = promotionDao.getByItemId(item.getId());
        item.setItemStock(itemStock);
        if (promotion != null && promotion.getStatus() == 0) {
            item.setPromotion(promotion);
        }
        // 回填本地缓存
        guavaCache.put("item:" + itemId, item);
        // 回填分布式缓存
        redisTemplate.opsForValue().set("item:" + itemId, item, 3 + new Random().nextInt(8), TimeUnit.MINUTES);
        return item;
    }

    @Override
    public Item getByIdFromCache(Integer itemId) {
        // 从本地缓存取
        Item item = (Item) guavaCache.getIfPresent("item:" + itemId);
        if (item != null) {
            return item;
        }
        // 从分布式缓存取
        item = (Item) redisTemplate.opsForValue().get("item:" + itemId);
        if (item != null) {
            // 回填本地缓存
            guavaCache.put("item:" + itemId, item);
        }
        return item;
    }

    @Override
    public void asyncIncreaseSales(int itemId, int amount) {
        JSONObject body = new JSONObject();
        body.put("itemId", itemId);
        body.put("amount", amount);
        Message<String> message = MessageBuilder.withPayload(body.toString()).build();
        rocketMQTemplate.asyncSend("seckill:increase_sales", message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                logger.info("商品[" + itemId + "]增加" + amount + "个销量成功.");
            }

            @Override
            public void onException(Throwable throwable) {
                logger.error("商品[" + itemId + "]增加" + amount + "个销量失败.");
            }
        }, 60 * 1000);
    }

    @Override
    public boolean increaseSales(int itemId, int amount) {
        int count = itemDao.increaseSales(itemId, amount);
        return count > 0;
    }

}
